package yxy.apple.util.concurrent;

import java.util.LinkedList;
import java.util.concurrent.*;

import yxy.apple.util.PubSub.Distributor;
import yxy.apple.util.PubSub.IPublisher;
import yxy.apple.util.PubSub.ISubscriber;
import yxy.apple.util.PubSub.ISubscription;

public class ThreadPoolManager implements IPublisher {

	static private ThreadPoolManager instance;

	public static ThreadPoolManager getInstance() {
		if (instance == null) {
			instance = new ThreadPoolManager();
		}
		return instance;

	}

	private ThreadPoolManager() {
		init();
	}

	private static final long MAX_DELAY = TimeUnit.NANOSECONDS.toMillis(Long.MAX_VALUE - System.nanoTime()) / 2;

	private Distributor distributor;

	private ScheduledThreadPoolExecutor scheduledPool;
	private ThreadPoolExecutor instantPool;
	private ThreadPoolExecutor onePool;

	private RejectedExecutionHandler handler;

	private static ThreadConfig config;

	private void init() {
		createDistributer();
	}

	private void createDistributer() {
		// 创建分销器
		distributor = new Distributor(new LinkedList<ISubscriber>()) {

			@Override
			public void notify(ISubscription subscription) {
				for (ISubscriber subscriber : collection()) {
					try {
						subscriber.receive(subscription);
					} catch (Exception e) {
					}
				}
			}
		};
	}

	@Override
	public void subscribe(ISubscriber subscriber) {
		distributor.register(subscriber);
	}

	@Override
	public void publish(ISubscription subscription) {
		distributor.notify(subscription);
	}

	/** 启动线程池 */
	public void start(ThreadConfig config) {
		ThreadPoolManager.config = config;

		handler = new RejectedExecutionHandler() {
			@Override
			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
				if (executor.isShutdown()) {
					return;
				}
				publish(new ThreadPoolManagerLogger("线程池s%饱和，额外执行s%", executor.toString(), r.toString()));
				if (Thread.currentThread().getPriority() > Thread.NORM_PRIORITY) {
					new Thread(r).start();
				} else {
					r.run();
				}
			}
		};

		final int instantPoolSize = Math.max(1, config.getThreadPoolSize() / 2);
		scheduledPool = new ScheduledThreadPoolExecutor(config.getThreadPoolSize() - instantPoolSize);
		scheduledPool.setRejectedExecutionHandler(handler);
		scheduledPool.prestartAllCoreThreads();

		instantPool = new ThreadPoolExecutor(instantPoolSize, 800, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		instantPool.setRejectedExecutionHandler(handler);
		instantPool.prestartAllCoreThreads();

		// 20以内性能最好，队列长度1
		onePool = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
		onePool.setRejectedExecutionHandler(handler);
		onePool.prestartAllCoreThreads();

		scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				purge();
			}
		}, 60000, 60000);

		publish(new ThreadPoolManagerLogger("线程池已启动！启动线程数：%d : %d", instantPoolSize, config.getThreadPoolSize()));
	}

	/** 关闭线程池 */
	public void shutdown() {
		publish(new ThreadPoolManagerLogger("ThreadPoolManager: 关闭，执行中的任务：[scheduledPool: d%]\t[instantPool: d%]\t[onePool: d%]", getTaskCount(scheduledPool),
				getTaskCount(instantPool), getTaskCount(onePool)));

		final long begin = System.currentTimeMillis();

		scheduledPool.shutdown();
		instantPool.shutdown();
		onePool.shutdown();

		boolean success = false;
		try {
			success |= awaitTermination(5000);

			scheduledPool.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
			scheduledPool.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);

			success |= awaitTermination(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		if (success) {
			publish(new ThreadPoolManagerLogger("ThreadPoolManager: 关闭完成，耗时 d% 毫秒，剩余任务：[scheduledPool: d%]\t[instantPool: d%]\t[onePool: d%]",
					System.currentTimeMillis() - begin, getTaskCount(scheduledPool), +getTaskCount(instantPool), getTaskCount(onePool)));
		} else {
			publish(new ThreadPoolManagerLogger("ThreadPoolManager: 关闭未完成，耗时 d% 毫秒，剩余任务：[scheduledPool: d%]\t[instantPool: d%]\t[onePool: d%]",
					System.currentTimeMillis() - begin, getTaskCount(scheduledPool), +getTaskCount(instantPool), getTaskCount(onePool)));
		}
	}

	public void purge() {
		scheduledPool.purge();
		instantPool.purge();
		onePool.purge();
	}

	private int getTaskCount(ThreadPoolExecutor tp) {
		return tp.getQueue().size() + tp.getActiveCount();
	}

	private boolean awaitTermination(long timeoutInMillisec) throws InterruptedException {
		final long begin = System.currentTimeMillis();

		while (System.currentTimeMillis() - begin < timeoutInMillisec) {
			if (!scheduledPool.awaitTermination(10, TimeUnit.MILLISECONDS) && scheduledPool.getActiveCount() > 0) {
				continue;
			}
			if (!instantPool.awaitTermination(10, TimeUnit.MILLISECONDS) && instantPool.getActiveCount() > 0) {
				continue;
			}
			if (!onePool.awaitTermination(10, TimeUnit.MILLISECONDS) && onePool.getActiveCount() > 0) {
				continue;
			}
			return true;
		}

		return false;
	}

	private final long validate(long delay) {
		return Math.max(0, Math.min(MAX_DELAY, delay));
	}

	private static final class ThreadPoolExecuteWrapper extends ExecuteWrapper {
		private ThreadPoolExecuteWrapper(Runnable runnable) {
			super(runnable);
		}

		@Override
		protected long getMaximumRuntimeInMillisecWithoutWarning() {
			return ThreadPoolManager.config.MaxRuntimeMillisecWithoutWarning;
		}
	}

	// ===========================================================================================

	/**
	 * once 不循环
	 * 
	 * @param r
	 * @param delay
	 * @return
	 */
	public final ScheduledFuture<?> schedule(Runnable r, long delay) {
		r = new ThreadPoolExecuteWrapper(r);
		delay = validate(delay);

		return new ScheduledFutureWrapper(scheduledPool.schedule(r, delay, TimeUnit.MILLISECONDS));
	}

	// ===========================================================================================

	/**
	 *
	 * @param r
	 * @param delay  延时运行 毫秒
	 * @param period 休眠时间 以运行后节点休眠
	 * @return
	 */
	public final ScheduledFuture<?> scheduleAtFixedRate(Runnable r, long delay, long period) {
		r = new ThreadPoolExecuteWrapper(r);
		delay = validate(delay);
		period = validate(period);

		return new ScheduledFutureWrapper(scheduledPool.scheduleAtFixedRate(r, delay, period, TimeUnit.MILLISECONDS));
	}

	/**
	 *
	 * @param r
	 * @param delay
	 * @param period 以运行开始节点休眠
	 * @return
	 */
	public final ScheduledFuture<?> scheduleWithFixedDealy(Runnable r, long delay, long period) {
		r = new ThreadPoolExecuteWrapper(r);
		delay = validate(delay);
		period = validate(period);

		return new ScheduledFutureWrapper(scheduledPool.scheduleWithFixedDelay(r, delay, period, TimeUnit.MILLISECONDS));
	}
	// ===========================================================================================

	public final void execute(Runnable r) {
		r = new ThreadPoolExecuteWrapper(r);

		instantPool.execute(r);
	}

	public final void executeOrderRunning(Runnable r) {
		r = new ExecuteWrapper(r);

		onePool.execute(r);
	}

	// ===========================================================================================

	public final Future<?> submit(Runnable r) {
		r = new ThreadPoolExecuteWrapper(r);

		return instantPool.submit(r);
	}

	public final Future<?> submitLongRunning(Runnable r) {
		r = new ExecuteWrapper(r);

		return onePool.submit(r);
	}

	public final <T> Future<T> submit(Runnable r, T result) {
		r = new ThreadPoolExecuteWrapper(r);
		return instantPool.submit(r, result);
	}

	// ===========================================================================================
}
